import beans.OrderEvent;
import beans.ReceiptEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author zkq
 * @date 2022/10/16 15:30
 */
public class TxPayMatchByJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取订单支付事件数据
        SingleOutputStreamOperator<OrderEvent> orderStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\OrderLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new OrderEvent(new Long(split[0]), split[1], split[2], new Long(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                )
                .filter(data -> !"".equals(data.getTxId())); //交易id不为空,pay事件

        //读取到账事件数据
        SingleOutputStreamOperator<ReceiptEvent> receiptStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\ReceiptLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new ReceiptEvent(split[0], split[1], new Long(split[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ReceiptEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEvent>() {
                            @Override
                            public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                );

        //区间连接两条流得到匹配的数据
        SingleOutputStreamOperator<Tuple2<OrderEvent,ReceiptEvent>> resultStream = orderStream
                .keyBy(OrderEvent::getTxId)
                .intervalJoin(receiptStream.keyBy(ReceiptEvent::getTxId))
                .between(Time.seconds(-3), Time.seconds(5))
                .process(new TxPayMatchDetectByJoin());

        resultStream.print("matched");

        env.execute();
    }
        //此方式只能拿到匹配上的 相比于connect差点意思 connect可以拿到匹配不上的进行侧输出
        //这里虽然也可以侧输出 但是进入的数据都是连接上的数据 拿不到连接不上的
    public static class TxPayMatchDetectByJoin extends ProcessJoinFunction<OrderEvent,ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>>{

        @Override
        public void processElement(OrderEvent left, ReceiptEvent right, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            out.collect(new Tuple2<>(left,right));
        }
    }
}
